package com.siebre.picclife.asynctask.model.impl;

import static com.google.common.collect.Iterables.tryFind;

import java.io.IOException;
import java.util.Date;

import javax.persistence.Column;
import javax.persistence.Entity;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Duration;
import org.joda.time.Instant;

import com.google.common.base.Functions;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.base.Throwables;
import com.google.common.collect.FluentIterable;
import com.siebre.bmf.annotation.BmfClazz;
import com.siebre.bmf.annotation.BmfProperty;
import com.siebre.bmo.BmoImpl;
import com.siebre.picclife.asynctask.model.Task;
import com.siebre.picclife.asynctask.model.TaskStatus;

/**
 * 
 * @author ZhangChi
 * @since 2015年4月3日
 */
@Entity
@BmfClazz
public class TaskEntity extends BmoImpl implements Task {

	private static final Joiner JOINER = Joiner.on(IOUtils.LINE_SEPARATOR);

	private String type;
	
	private String argument;
	
	private TaskStatus status;

	private int failureCount;
	
	private String processor;

	private Date startTime;
	
	private long processTime;

	@Column(length = 1000)
	private String resultMessage;
	
	public TaskEntity(String type, String argument) {
		this.type = type;
		this.argument = Preconditions.checkNotNull(argument, "argument can not be null");
		status = TaskStatus.NEW;
	}

	/**
	 * used by framework
	 */
	protected TaskEntity() {
	}

	@Deprecated
	public int start() {
		return startBy("default");
	}

	public int startBy(String processor) {
		Preconditions.checkNotNull(processor, "processor can not be null");
		Preconditions.checkState(status.equals(TaskStatus.NEW) || status.equals(TaskStatus.FAILED), 
				String.format("task in %s status can not be started", status));
		status = TaskStatus.PROCESSING;
		startTime = Instant.now().toDate();
		this.processor = processor;
		return failureCount + 1;
	}

	public void complete() {
		completeWith(TaskStatus.SUCCEEDED, null);
	}

	public TaskStatus complete(Throwable cause) {
		TaskStatus toStatus = tryFind(Throwables.getCausalChain(cause), Predicates.instanceOf(IOException.class)).isPresent() ? 
				TaskStatus.FAILED : TaskStatus.CORRUPTED;
		Iterable<String> messages = FluentIterable
				.from(Throwables.getCausalChain(cause))
				.transform(Functions.toStringFunction());
		String message = JOINER.join(messages);
		completeWith(toStatus, message);
		return status;
	}

	public TaskStatus completeWithFailure(String message) {
		completeWith(TaskStatus.FAILED, message);
		return status;
	}

	private void completeWith(TaskStatus toStatus, String message) {
		Preconditions.checkState(status.equals(TaskStatus.PROCESSING), String.format("task in %s can not be completed", status));
		status = toStatus;
		if (toStatus.equals(TaskStatus.FAILED) || toStatus.equals(TaskStatus.CORRUPTED)) {
			failureCount++;
		}
		processTime = new Duration(getStartInstant(), Instant.now()).getMillis();
		resultMessage = StringUtils.abbreviate(message, 1000);
	}
	
	public void markAsCorrupted() {
		status = TaskStatus.CORRUPTED;
	}

	public String getType() {
		return type;
	}

	public String getArgument() {
		return argument;
	}

	@BmfProperty(modelFeature = false)
	public TaskStatus getStatus() {
		return status;
	}

	public int getFailureCount() {
		return failureCount;
	}

	public ProcessStats lastProcessStats() {
		return new ProcessStats(processor, getStartInstant(), getProcessDuration(), resultMessage);
	}

	public String getProcessor() {
		return processor;
	}
	
	@BmfProperty(modelFeature = false)
	public Instant getStartInstant() {
		return new Instant(startTime);
	}

	/**
	 * for BmfClass feature
	 * 
	 * @return
	 * @deprecated use {@link #getStartInstant()}
	 */
	public Date getStartTime() {
		return startTime;
	}
	
	@BmfProperty(modelFeature = false)
	public Duration getProcessDuration() {
		return Duration.millis(processTime);
	}

	/**
	 * for BmfClass feature
	 * 
	 * @return
	 * @deprecated use {@link #getProcessDuration()}
	 */
	public long getProcessTime() {
		return processTime;
	}

	public String getResultMessage() {
		return resultMessage;
	}

	@Override
	public String toString() {
		return "TaskEntity [type=" + type + ", argument=" + argument + "]";
	}
}
